package virtualinit;

import virtualinit.util.CommonUtil;
import virtualinit.util.Log;
import virtualinit.util.StopWatch;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.concurrent.*;

public class InsertTask implements Runnable {
    Config config = Config.instance;
    static Log log = Log.instance;
    private volatile boolean errorOccur = false;

    private static String DUPLICATE_ENTRY = "Duplicate entry";
    private ThreadPoolExecutor pool;
    private StopWatch watch;
    private ArrayBlockingQueue<Task> sqlQueue;
    private String taskName;
    private volatile boolean hasSource = true;
    int batchSize;

    Thread currentThread;

    public boolean canExit() {
        return errorOccur || (sqlQueue.size() == 0 && lastTaskCompleted && !hasSource);
    }

    @Override
    public void run() {
        currentThread = Thread.currentThread();
        log.info("Insert task start...");
        while (!canExit()) {
            try {
                Task task = sqlQueue.take();
                pool.execute(() -> executeInsert(task));
            } catch (InterruptedException e) {
                log.info("task:{} hasSource:{} errOccur:{} lastTaskCompleted:{} 被Interrupted,任务完成",
                        taskName, hasSource, errorOccur, lastTaskCompleted);
                return;
            }
        }
    }

    public void init(AbstractInit init) {
        this.sqlQueue = new ArrayBlockingQueue<>(config.getInteger("config.sqlHolderQueueSize", 10));
        this.batchSize = config.batchSize;
        this.watch = new StopWatch(sqlQueue);
        this.taskName = init.getTaskName();
        this.pool = new ThreadPoolExecutor(config.threadPoolSize, config.threadPoolSize,
                600L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(500), Executors.defaultThreadFactory(), (r, e) -> {
            try {
                e.getQueue().put(r);
            } catch (Exception e12) {
                e12.printStackTrace();
            }
        });
        new Thread(this).start();
        log.info("初始化InsertTask, threadPool size={}, tipsSize={} sqlQueueSize:{} taskName={}",
                config.threadPoolSize, watch.tipsSize, sqlQueue.remainingCapacity(), taskName);
    }

    volatile boolean lastTaskCompleted = true;
    long duplicateCount = 0;

    public int executeInsert(Task task) {
        long start = System.currentTimeMillis();
        lastTaskCompleted = false;
        PreparedStatement prep = null;
        Connection connection = null;
        int update = 0;
        try {
            connection = task.ds.takeConnection();
            prep = connection.prepareStatement(task.sql);
            update = prep.executeUpdate();
        } catch (Exception e) {
            if (e.getMessage().contains(DUPLICATE_ENTRY)) {
                try {
                    if (duplicateCount++ % 10000 == 0)
                        log.error("Duplicate task=" + substringSql(task.sql));
                    //                    sqlFailQueue.put(task);
                } catch (Exception ex) {
                    log.error("插入出错：" + task, e);
                    errorOccur = true;
                    throw new RuntimeException("插入出错：" + substringSql(task.sql), ex);
                }
            } else {
                errorOccur = true;
                throw new RuntimeException("插入出错,msg=" + e.getMessage() + "\ntask=" + substringSql(task.sql), e);
            }
        } finally {
            watch.record(update, start);
            lastTaskCompleted = true;
            CommonUtil.closeQuietly(prep);
            task.ds.returnConnection(connection);
        }
        return update;
    }

    private String substringSql(String sql) {
        return sql.length() > 1000 ? sql.substring(0, 1000) : sql;
    }

    public void addSql(SqlHolder sqlHolder) {
        if (sqlHolder.hasSql()) {
            try {
                sqlQueue.put(new Task(sqlHolder.getSql(), sqlHolder.ds));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void handleCompleted() {
        hasSource = false;
        while (true) {
            if (canExit()) {
                currentThread.interrupt();
                break;
            }
            CommonUtil.sleep(500);
        }
        pool.shutdown();
    }

    public static class Task {
        String sql;
        Ds ds;

        public Task(String sql, Ds ds) {
            this.sql = sql;
            this.ds = ds;
        }
    }

    public StopWatch getWatch() {
        return watch;
    }
}
